package gbench.sandbox.matlib;

import org.junit.jupiter.api.Test;

import java.util.ArrayList;
import java.util.Comparator;
import java.util.stream.Collector;
import java.util.stream.Collectors;
import java.util.stream.Stream;
import java.time.LocalDateTime;

import gbench.common.fs.FileSystem;
import gbench.common.matlib.MatlibCanvas;
import gbench.common.matlib.MatlibCanvas.IVPair;
import gbench.common.matlib.MatlibCanvas.NVec;
import gbench.common.matlib.MatlibCanvas.Range;
import gbench.common.matlib.MatlibCanvas.Seq;
import gbench.common.matlib.MatlibCanvas.XVec;
import gbench.common.matlib.data.DataReader;
import gbench.common.matlib.data.DataReader.DFrame;
import gbench.common.matlib.rdd.BasePair;
import gbench.common.matlib.rdd.PVec;
import gbench.common.tree.LittleTree.IRecord;

import static gbench.common.tree.LittleTree.KVPair.kvpclc;
import static gbench.common.tree.LittleTree.IRecord.A;
import static gbench.common.tree.LittleTree.IRecord.r2uclc2;
import static gbench.common.matlib.data.DataReader.DFrame.dfmclc;
import static gbench.common.matlib.data.DataReader.DFrame.bp2kvpf1;
import static gbench.common.matlib.MatlibCanvas.Weightable.WEIGHT;
import static gbench.common.matlib.MatlibCanvas.println;
import static gbench.common.matlib.rdd.BasePair.BP;
import static gbench.common.matlib.rdd.PVec.pvec1clc;
import static gbench.common.matlib.rdd.PVec.xkv2pvec;
import static gbench.common.tree.LittleTree.IRecord.rb;
import static java.util.Comparator.comparing;


/**
 * PVec 的使用演示 <br>
 * <br>
 * PVec 的基础结构是 BasePair (BP), BP的K,V都可以是任意类型。是 BP&lt;K,V&gt; <br>
 * IRecord 的基础结构是 KVPair (KVP) ,KVP 的K必须是字符串类型,V 可以是任意类型。是 KVP&lt;String,V&gt; <br>
 * DFrame 本质是一个特殊的IRecord,基础也是KVPair (KVP),除了 K必须是 字符串类型以外,V 需要是一个向量类型<br>
 * 即 Collection,Array,或是 TVec 类型等集合类型。是KVP(String,[V]&gt; <br>
 * <br>
 * 所以 PVec,IRecord,DFrame 三者是同构的,但是 基础 不同:<br>
 * PVec 是 BP&lt;K,V&gt; 向量 <br>
 * IRecord 是 KVP&lt;String,V&gt; 向量 <br>
 * DFrame 是 KVP&lt;String,[V]&gt; 向量 <br>
 * 
 * @author gbench
 *
 */
public class JunitPVec {
    
    @Test
    public void foo(){
        final var x = NVec.of(10,i->i);
        println("pvec groupByKey:",x.mutateX(xvec->xvec.collect(pvec1clc(e->e%2==0))).groupByKey());
        println("pvec groupByKey shape",x.mutateX(xvec->xvec.collect(pvec1clc(e->e%2==0))).groupByKey().shape());
        println("pvec groupByKey summarizingDouble",x.mutateX(xvec->xvec.collect(pvec1clc(e->e%2==0))).reduceByKey(Collectors.summarizingDouble(e->e)));
        println("pvec reduceByValue toList",x.mutateX(xvec->xvec.collect(pvec1clc(e->e%2==0))).reduceByValue(Collectors.toList()));
        println("PVec的 各种类型变换: PVec=>IRecord=>DFrame,其实他们之间是等价的  PVec<=>IRecord<=>DFrame，只是各自依赖的基础单元不同,是同构概念。\n"
                + "- PVec 的元素类型 是 : BP<K,V> 键-值-类型都不约束\n"
                + "- IRecord 的元素类型 是 : KVP<String,V> 键类型约束-值类型都不约束\n"
                + "- DFrame 的元素类型 是 : KVP<String,[V]> 键类型约束-值类型约束\n",
                x.mutateX(xvec->xvec.collect(pvec1clc(e->e%2==0)))
                .reduceByKey(Collectors.toList()) // 按照key进行分组
                .mutateX(pvec->pvec.map(DFrame::bp2kvp).map(kvp->kvp.fmap1(Object::toString)).collect(kvpclc(IRecord::KVS2REC))) // 转换成 IRecord
                .mutate(DFrame::new));
        
        final var pvec = x.fmap(BasePair::bp).mutateX(xkv2pvec());
        println("\npvec groupByKey",pvec.groupByKey());
        println("pvec car",pvec.car());
        println("pvec cdrX",pvec.cdrX());
        println("shuffle weight",pvec.shuffleX((i,e)-> WEIGHT(e)));
        println("shuffle",pvec.shuffle());
        println("x fmap kvp:",x.fmap(BasePair::bp).collect(PVec.pvec2clc(e->e.fmap(p->p*4))));
        
        final var words = XVec.OF("this is a good story and i like it".split("")).fmap(e->BP(e, 1)).collect(PVec.pvec2clc());
        println("\nwords 词频统计 reduceByKey",words.reduceByKey(vv-> vv.size()).sorted(comparing(IVPair::_2)));
        println("words 词频统计 groupByKey",words.groupByKey().fmap2(e->e.size()).sorted(comparing(IVPair::_2)));
        println("words 词频统计 groupByKey fmap",words.groupByKey().fmap(p->p.fmap(v->v.size()).fmap1(k->k+":")).sorted(comparing(IVPair::_2)));
        println("words 词频统计 groupByKey fmap fswap",words.groupByKey().fmap(p->p.fmap(v->v.size()).fmap1(k->k+":")).sorted(comparing(IVPair::_2)).fswap());
        
        println("\nwords 交换键值fswap",words.fswap());
        println("words 交换键值mutateX fswap",words.mutateX(PVec::fswap));
        println("words groupByValue",words.groupByValue());
        println("words groupByValue sorted",words.groupByValue().fmap2(e->e.sorted()));
        println("words reduceByValue",words.reduceByValue(e->e.size()));
    }
    
    /**
     * 纵向分解
     */
    @Test
    public void split() {
        final var datafile = FileSystem.path("data.txt", this.getClass());
        final var iris = DataReader.loaddfs(datafile).get("iris");
        final var p1 = iris.shuffle().head(10)
                .split("Id,Sepal.Length,Sepal.Width", "Species")
                .unwind(dfmclc,dfmclc);
        println("iris p1:\n", p1._1());
        println("Species\n", p1._2().str(0));
        println("Species", p1._2().str("Species"));
        
        final var p2 = iris.shuffle().head(10)
                .split(Range.rG(0,4), Range.rG(5,5))
                .unwind(dfmclc,dfmclc);
        println("\niris p2:\n", p2._1());
        println("Species", p2._2().str("Species"));
        
        final var p3 = iris.shuffle().head(10)
                .split(A(0,1,2,3,4), A(5))
                .unwind(dfmclc,dfmclc);
        println("\n iris p3:\n", p3._1());
        println("Species", p3._2().str("Species"));
    }
    
    /**
     * BasePair Builder 的使用演示
     */
    @Test
    public void bpb(){
        final var id = 1;
        final var name = "张三";
        final var bpb = BasePair.BPB(id); // BasePair Builder
        println("bpb 生成对象",bpb.join(name),bpb.prepend(name));
        
        final var rec = NVec.of(10,i->i)
                .collect(PVec.pvec1clc(i->i%2==0))
                .map(DFrame::bp2kvp).map(e->e.swap().fmap1(Object::toString))
                .collect(IRecord.rclc2);// pvec -> rec
        println("pvec 转换成 rec",rec);
        println("copy",NVec.of(10,i->i)
                .collect(PVec.pvec1clc(i->i%2==0))
                .copy(e->e.fmap2(p->p*2)));
    }
    
    /**
     * sorted 排序
     */
    @Test
    public void sorted() {
        final var pos = NVec.of(0, 2, 4, 7, 10);
        final Comparator<NVec> comparator = (a, b) -> Stream.of(a, b).map(e -> e.resize(2, NVec.of(0))).reduce(NVec::sub).get().stream()
                .dropWhile(e -> e == 0d).findFirst().map(Double::intValue).orElse(0);
        Range.cph(pos, pos)
                .map(NVec::sorted).sorted(comparator)
                .distinct().filter(e -> !e.get(0).equals(e.get(1))).map(e -> e.get(1) - e.get(0))
                .sorted()
                .forEach(MatlibCanvas::println);
        println(NVec.of(0,1));
    }
    
    /**
     * 内切酶的片段
     */
    @Test
    public void pdp() {
            final var pos = NVec.of(0, 2, 4, 7, 10);
            println("pdp:");
            Range.cph(pos, pos)
                    .map(NVec::sorted).sorted()
                    .distinct().filter(e -> !e.get(0).equals(e.get(1))).map(e -> e.get(1) - e.get(0))
                    .sorted()
                    .forEach(MatlibCanvas::println);
            
            println("compareTo:");
            println("NVec.of(5,1,null,5,6).compareTo(NVec.of(5,1,null,5))",NVec.of(5,1,null,5,6).compareTo(NVec.of(5,1,null,5)));
            println("XVec.OF(\"hello\").compareTo(XVec.OF(\"this\",\"you\"))",XVec.OF("hello").compareTo(XVec.OF("this","you")));
     
    }
    
    /**
     * 元素提取
     */
    @Test
    public void get() {
        final var x=NVec.of(100,i->i).stream().collect(PVec.pvec1clc(e->e%2==0));
        println("pvec",x);
        final var p = x.groupByKey();
        final var df = p.map(bp2kvpf1(Object::toString)).collect(r2uclc2(DFrame::new));
        println("df\n",df);
        println("p.pget(true)",p.pget(true));
        println("p.pget(true,t->t.car())",p.pget(true,t->t.car()));
        println("p.pgetFirst(true)",p.pgetFirst(true));
        println("p.pgetFirst(true).map(t->t.value().car())",p.pgetFirst(true).map(t->t.value().car()));
    }
    
    /**
     * BasePair的扁平化处理
     */
    @Test
    public void flatten(){
        final var xx = NVec.of(100,i->i);
        final var pvec = xx.collect(PVec.pvec1clc(e->e%2==0));
        final var df = pvec.map(e->e.swap().append("zhang"+e._2().intValue()).append(LocalDateTime.now().minusDays((int)(Math.random()*10000)))) // 数据扁平化
                .map(bp->bp.flatten(rb("id,sex,name,birth")::build)).sorted(IRecord.cmp("birth",false)).collect(dfmclc);
        println(df);
    }
    
    /**
     * 多项式乘法
     */
    @Test
    public void ploynomials_mul() {
        final var pn = NVec.of(1, 2, 3); // 多项式
        final var line=pn.flatMap((i, a) -> pn.map((j, b) -> BasePair.bp(i + j, a * b)))
        .collect(PVec.pvec2clc()).groupByKey()
        .map(e -> {
            println(e);
            return e._2().reduce2((a, b) -> a + b) + "*x^" + e._1();
        }).collect(Collectors.joining("+"));
        println("\n",line);
        println(pn.collect(pmulclc(pn)));
        
        final var px = NVec.of(1,1);
        final var tt = px
            .collect(pmulclc(px)) // 平方
            .collect(pmulclc(px)) // 立方
            .collect(pmulclc(px)); // 四次方
        
        println("(1+1)^4", tt, tt.reduce2((a, b) -> a + b));
    }

    /**
     * 键值对儿的快速生成。
     */
    @Test
    public void kvp() {
        final var pvec = PVec.of(1,2,3,4,5,6,7,8).cast(Integer.class, Integer.class);
        println(pvec);
        try {
            final var pvec2_except = PVec.of(1,2,3,4,5,6,7).cast(Integer.class, Integer.class);// 因为元素数量是奇数所以报错了。
            println(pvec2_except);
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    /**
     * 多项式的乘法的归集器
     * 
     * @param <T> 多项式类型
     * @param ploynomials  多项式表达式
     * @return 多项式的乘法
     */
    @SuppressWarnings("unchecked")
    public static <T extends Number> Collector<T, ?, Seq<T>> pmulclc(final Seq<T> ploynomials) {
        return Collector.of(() -> new ArrayList<T>(), ArrayList::add, (a, b) -> {a.addAll(b); return a;}, aa -> { // 多项式乘法
            final var bp = XVec.OF(aa) // 把一个 数组列表 转换成 XVec
                .flatMap((i, a) -> ploynomials.map((j, b) -> BP(i + j, a.doubleValue() * b.doubleValue()))) // 分配率
                .collect(PVec.pvec2clc()) // 合并成PVec
                .reduceByKey(xvec -> xvec.reduce2((a, b) -> a + b)) // 合并同次幂的项目，合并同类项
                .sorted(e -> e._1()) // 按照幂次从低到高进行排序
                .unwind(); // 拆解成幂次向量 和 系数向量
            return (Seq<T>) bp._2(); // 提取系数向量
        });
    }
}
